package com.shujia.traffic.dws

import org.apache.spark.internal.Logging
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.slf4j.Logger

object DwsAppAcdMrjqDay extends Logging{

  val logger: Logger = log
  // 每日警情统计
  def main(args: Array[String]): Unit = {

    // 接收分区日期参数
    if(args.length==0){
      logger.error("请指定日期分区参数")
      return
    }

    val day: String = args.head
    logger.info(s"当前正在处理的分区为：$day")

    // 创建SparkSession环境
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppAcdMrjqDay")
      .config("spark.sql.shuffle.partitions", "4")
      .enableHiveSupport() // 开启Hive的支持
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 构建DF
    // 从Hive获取dwd的表
    val sgDF: DataFrame = spark
      .table("dwd.dwd_base_acd_file_msk_f_d")
      .where($"ds" === day)

    // 统计每日的事故数、死亡事故数、死亡人数、受伤人数、财产损失
    var drCntDF: DataFrame = sgDF
      .groupBy(substring($"sgfssj", 1, 10) as "tjrq")
      .agg(
        // 当日事故数
        count($"sgbh") as "dr_sgs"
        // 当日死亡事故数
        , sum(when($"swrs24" + $"swrs3" + $"swrs7" + $"swrs30" + $"swrsq" > 0, 1).otherwise(0)) as "dr_swsgs"
        , sum($"swrs30") as "dr_swrs"
        , sum($"ssrs30") as "dr_ssrs"
        , sum($"zjccss") as "dr_zjccss"
      )
      .where($"tjrq".rlike("\\d{4}-\\d{2}-\\d{2}"))


    // 需要统计的指标类型列表
    val indexList: List[String] = List[String]("sgs", "swsgs", "swrs", "ssrs", "zjccss")
    // 分别计算当年总数及同比
    for (index <- indexList) {
      drCntDF = drCntDF
        .withColumn(s"jn_$index", sum($"dr_$index") over Window.partitionBy(year($"tjrq")))
        .withColumn(s"last_dr_$index", lag($"dr_$index", 1, 0) over Window.partitionBy(substring($"tjrq", 6, 5)).orderBy(year($"tjrq")))
        .withColumn(s"tb_$index", coalesce(round($"dr_$index" * 100 / $"last_dr_$index" - 100, 4), expr("0")))
    }

    drCntDF
      // 整理数据
      .select(
        $"tjrq"
        , $"dr_sgs"
        , $"jn_sgs"
        , $"tb_sgs"
        , $"dr_swsgs"
        , $"jn_swsgs"
        , $"tb_swsgs"
        , $"dr_swrs"
        , $"jn_swrs"
        , $"tb_swrs"
        , $"dr_ssrs"
        , $"jn_ssrs"
        , $"tb_ssrs"
        , $"dr_zjccss"
        , $"jn_zjccss"
        , $"tb_zjccss"
      )
      // 保存数据到dws表的目录中
      .write
      .mode(SaveMode.Overwrite)
      .format("csv")
      .option("sep", ",")
      .save(s"/daas/motl/traffic/dws/dws_app_acd_mrjq_d/ds=$day")

    // 增加分区
    spark.sql(
      s"""
        |alter table dws.dws_app_acd_mrjq_d add if not exists partition(ds='$day')
        |""".stripMargin)

  }

}
